package hz

/**
  * Created by hz on 2017/1/6.
  */


  import java.text.SimpleDateFormat
  import java.util
  import java.util.Date

  import kafka.serializer.StringDecoder
  import org.apache.log4j.{Level, Logger}
  import org.apache.spark.SparkConf
  import org.apache.spark.rdd.RDD
  import org.apache.spark.streaming.dstream.{DStream, InputDStream}
  import org.apache.spark.streaming.kafka.KafkaUtils
  import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}
  import org.elasticsearch.spark._

  import scala.util.control._
  import scala.collection.mutable.ArrayBuffer



  object Kafkauu3{

    def main(args:Array[String]){
      Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
      Setting.readProperties("default1.properties")
      val sparkConf = new SparkConf().set("spark.cores.max","1")//.setAppName("test")
      //val sparkConf = new SparkConf().setMaster("local[2]").setAppName("test")
      sparkConf.set("es.nodes", Setting.getProperty("esnode"))//spark://hz:7077  ElasticSearch
      sparkConf.set("es.port", Setting.getProperty("esport"))
      sparkConf.set("es.index.auto.create", "true")
      val scc = new StreamingContext(sparkConf, Duration(5000))
      val topics = Set(Setting.getProperty("topic"))
      val kafkaParam = Map(
        "metadata.broker.list" -> Setting.getProperty("brokerlist") // kafka的broker list地址
      )
      val b=new SimpleDateFormat("yyyy-MM-dd")
      val stream: InputDStream[(String, String)] = createStream(scc, kafkaParam, topics)
      val lei = stream.map(_._2)
      lei.persist()

      lei.print()
      val rule=DBUtil.getRule(args(0))
      val min1:Int=min(rule);
      val h = lei.window(Seconds(min1), Seconds(5))

      if (rule.size() == 0) {
        lei.print()
      } else {
        val loop=new Breaks;
        lei.print()
        var rs=h.foreachRDD(rdd => {
          loop.breakable{
            for (i <- 0 until rule.size()) {
              println("普通规则")
              val rs=zhixing(rdd, rule.get(i))
              if (rs.count()>0){
                rs.saveToEs("baojing1"+b.format(new Date())+"/guizefenlei")
              }
            }
            if (rule.size()==1){
              println("场景1")
              var rs=zhixing(rdd,rule.get(0),args(0))
              if (rs.count()>0){
                rs.saveToEs("baojingchangjing"+b.format(new Date())+"/guizefenlei")
              }
            }
            if (rule.size()>1){
              println("场景2")
              var i=0;
              var rs=zhixing(rdd,rule.get(0),args(0))
              while(rs.count()>0 && i<rule.size()){
                rs=zhixing1(rs,rule.get(i),args(0))
                i+=1
              }
              if (rs.count()>0 && i==rule.size()-1){
                rs.saveToEs("baojingchangjing"+b.format(new Date())+"/guizefenlei")
              }
            }
          }
        })


      }


      scc.start() // 真正启动程序
      scc.awaitTermination() //阻塞等待
    }


    def min(rules:util.List[RuleBean]):Int={
      var min=rules.get(0).getTimerange;
      for (i <-0 until rules.size()){
        if (min>rules.get(i).getTimerange) min=rules.get(i).getTimerange
      }
      min
    }



    def zhixing(lei:RDD[String],rule:RuleBean,cj:String=""):RDD[Map[String,String]]={
      val leixing:String=DBUtil.getLeixing
      var rizhi1:RDD[Map[String,String]]=null

      println("这是第"+rule.getId+"")
      val rizhi:RDD[Map[String,String]]=lei.map(x => denghao(x)).map(line => jiexi(line, rule.getId,cj)).filter(_.contains(leixing))

      rizhi1=tiaojian(rizhi,rule,cj)

      val rs=paixu1(rizhi1,rule)

      rs
    }


    //场景
    def zhixing1(lei:RDD[Map[String,String]],rule:RuleBean,cj:String=""):RDD[Map[String,String]]={

      var rizhi1:RDD[Map[String,String]]=null
      rizhi1=tiaojian(lei,rule,cj)

      val rs=paixu1(rizhi1,rule)

      rs
    }

    /**
      * 进行条件匹配
      *
      *
      * @param rule1
      * @return
      */
    def tiaojian(rizhi:RDD[Map[String,String]],rule1:RuleBean,cj:String):RDD[Map[String,String]]={
      var rizhi1=rizhi

      val size=rule1.getTiaojian().size()
      val tiaojian= rule1.getTiaojian();
      if (tiaojian.size()!=0){
        for (i <- 0 until size) {
          val ziduan=tiaojian.get(i).getZiduan
          val zhi=tiaojian.get(i).getZhi
          tiaojian.get(i).getCaozuofu match {
            case "=" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 == zhi).map(x => x._1)
            case "<" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 < zhi).map(x => x._1)
            case ">" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 > zhi).map(x => x._1)
            case "<=" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 <= zhi).map(x => x._1)
            case ">=" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 >= zhi).map(x => x._1)
            case "!=" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 != zhi).map(x => x._1)
            case "Contains" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2.contains(zhi)).map(x => x._1)
            case "StartWith" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2.startsWith(zhi)).map(x => x._1)
            case "EndWith" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2.endsWith(zhi)).map(x => x._1)
          }
        }
      }
      rizhi1
    }


    def denghao(list:String):Array[String]={
      var tmp=""
      var ab=ArrayBuffer[String]()
      val length=list.length
      val exp="""([\w]+)=([\w]+[-(\+]?[\S]+)""".r
      val re=exp.findAllIn(list)
      for (i <- re){
        ab+=i
      }

      val exp1="""([\w]+)=\\\"([^\"]*)\\\"""".r
      val re1=exp1.findAllIn(list)
      for (j <- re1){
        print("woshijinlaile")
        ab+=j
      }
      val rab=ab.toArray
      for (i <- ab){
        print(i+" ")
      }
      return rab
    }


    /**
      * 进行排序
      *
      * @param list
      * @param rule
      */
    def paixu1(list:RDD[Map[String,String]],rule:RuleBean):RDD[Map[String,String]]={
      println("==进入排序==")
      var hh=list
      hh.cache()

      /**
        * 相同聚合时
        *
        * @param hello
        * @return
        */
      def zuhea(hello:Map[String,String]):String={
        val list=rule.getSamepol.split(",")
        var rs:String=""
        for (i <- list){
          rs+=hello(i)
        }
        rs
      }

      /**
        * 不同时聚合
        *
        * @param hello
        * @return
        */
      def zuheb(hello:Map[String,String]):String={
        val list=rule.getDiffpol.split(",")
        var rs:String=""
        for (i <- list){
          rs+=hello(i)
        }
        rs
      }




      var gv1:RDD[Iterable[Map[String,String]]]=null

      gv1=hh.map(x=>(zuhea(x),x)).groupByKey().map(_._2)
      val size=rule.getNumber

      val daxiao:RDD[Map[String,String]]= gv1.map(x=>(x,x.map(y=>zuheb(y)).toArray.distinct.size)).filter(_._2>=size).flatMap(x=>x._1.map(y=>y))  //不同时聚合




      /* if (daxiao.count() > 0)  // RDD has data
       {
         val b=new SimpleDateFormat("yyyy-MM-dd")
         println("存进去了")*/
      //daxiao.saveToEs("baojing"+b.format(new Date())+"/guizefenlei")
      //SpringUtil.getInstance().setbaojing()
      /* jingtai.baojing()
       val ip:String=daxiao.take(1).apply(0)("dst")
       val id=SpringUtil.getInstance().getasset(ip)

       val warn=new WarningBean();
       warn.setWarnname(rule.getRulename)
       warn.setAssetid(id)
       warn.setStatus("未清除")
       warn.setType("事件告警")
       warn.setRuntime((new java.util.Date().getTime-rule.getMktime.getTime).toInt)
       warn.setRuleid(rule.getId)
       //SpringUtil.getInstance().savebaojing(warn);
       warn11.save(warn)
       daxiao.count()*/

      //}
      daxiao
    }





    /**
      * 创建一个从kafka获取数据的流.
      *
      * @param scc           spark streaming上下文
      * @param kafkaParam    kafka相关配置
      * @param topics        需要消费的topic集合
      * @return
      */
    def createStream(scc: StreamingContext, kafkaParam: Map[String, String], topics: Set[String]) = {
      KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](scc, kafkaParam, topics)
    }

    /**
      * 将字符串解析为Map
      *
      * @param map
      * @param guize1
      * @return
      */
    def jiexi(map:Array[String],guize1:Int,cj:String):Map[String,String]={
      var map1:Map[String,String]=Map();
      var guize:Int=0;
      for (i <- map.indices){
        var cache1,cache2=""
        cache1=map(i).split("=")(0)
        cache2=map(i).split("=")(1)
        map1+=(cache1 -> cache2.replaceAll("""\\""","").replaceAll("\"",""))
      }
      val a=Array("dst","src")
      for(i <- 0 until a.size if !map1.contains(a(i))) map1+=(a(i) -> " ")
      map1+=("guize"->guize1.toString)
      if (cj!=""){ map1+=("cj"->cj)}

      map1
    }









}
